package com.sanji.kafka;

import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * Created by peter on 17-3-18.
 */
public class NativeProducer {
  public static void main(String[] args) {
    System.out.println("begin produce");
    for (int i = 0; i<=100;i++){
      connectionKafka("message:" + i);
    }
    try {
      Thread.sleep(2000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    System.out.println("finish produce");
  }

  public static void connectionKafka(String jsonStr) {
    System.out.println("aaaaa");
    Properties props = new Properties();
    props.put("bootstrap.servers", "192.168.2.44:32772");
    props.put("acks", "all");
    props.put("retries", 0);
    props.put("batch.size", 16384);
    props.put("linger.ms", 1);
    props.put("buffer.memory", 33554432);
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

    Producer<String, String> producer = new KafkaProducer<String,String>(props);

    if (StringUtils.isNotEmpty(jsonStr)) {
      System.out.println("begin produce");
      producer.send(new ProducerRecord<String, String>("topic", null, jsonStr));
      System.out.println("finish produce");
    }

  }

}
